package com.zidongxiangxi.sharding.job.core.node;

import com.zidongxiangxi.sharding.job.core.ShardingJobRegistry;
import com.zidongxiangxi.sharding.job.core.callback.LeaderExecutionCallback;
import com.zidongxiangxi.sharding.job.core.storage.NodeStorage;
import com.zidongxiangxi.sharding.job.core.util.BlockUtils;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.util.Objects;

/**
 * 主节点
 *
 * @author chenxudong
 * @date 2021/02/18
 */
@Slf4j
public class LeaderNode extends AbstractNode {
    private static final String ROOT = "leader";
    private static final String INSTANCE = ROOT + "/instance";
    private static final String LATCH = ROOT + "/latch";

    private final String leaderInstanceFullPath;

    public LeaderNode(final String jobName, final NodeStorage storage) {
        super(jobName, storage);
        this.leaderInstanceFullPath = getFullPath(INSTANCE);
    }

    public boolean isLeaderInstancePath(final String path) {
        return Objects.equals(leaderInstanceFullPath, path);
    }

    public void electLeader() {
        log.debug("Elect a new leader now.");
        storage.executeInLeader(LeaderNode.LATCH, new LeaderElectionExecutionCallback());
        log.debug("Leader election completed.");
    }

    public boolean isLeaderUntilBlock() {
        while (!hasLeader()) {
            log.info("选主中, 等待{}毫秒", 100);
            BlockUtils.waitingShortTime();
            electLeader();
        }
        return isLeader();
    }

    public boolean isLeader() {
        return ShardingJobRegistry.getInstance().getLocalInstanceId().equals(storage.getJobNodeData(INSTANCE));
    }

    public boolean hasLeader() {
        return storage.isJobNodeExisted(INSTANCE);
    }

    public void tryAbdication() {
        if (isLeader()) {
            storage.removeJobNodeIfExisted(INSTANCE);
        }
    }

    @RequiredArgsConstructor
    class LeaderElectionExecutionCallback implements LeaderExecutionCallback {

        @Override
        public void execute() {
            if (!hasLeader()) {
                storage.fillEphemeralJobNode(INSTANCE, ShardingJobRegistry.getInstance().getLocalInstanceId());
            }
        }
    }
}
